package com.uuah.server.transport.provider.mina.filter;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.uuah.server.UuahServer;

/**
 *
 * TODO comment ...
 *
 * <p>
 * update record:updator，update time，update content and version
 * </p>
 *
 * @author <a href="jonny_quan@hotmail.com">jonny</a>
 * @date 2009 2009-3-27 下午02:16:59
 * @version 1.0.0
 */
public class MinaExecutorFilter extends IoFilterAdapter {
	private transient final Logger logger = LoggerFactory.getLogger(getClass());

	private final Executor executor;

	public MinaExecutorFilter() {

		this(new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS,
				new LinkedBlockingQueue()));
	}

	public MinaExecutorFilter(Executor executor) {
		if (executor == null) {
			throw new NullPointerException("executor");
		}

		this.executor = executor;
	}

	public Executor getExecutor() {
		return executor;
	}

	private void fireEvent(NextFilter nextFilter, IoSession session,
			EventType type, Object data) {
		Event event = new Event(type, nextFilter, data);
		SessionBuffer buf = SessionBuffer.getSessionBuffer(session);

		boolean execute;
		synchronized (buf.eventQueue) {
			buf.eventQueue.offer(event);
			UuahServer.getServerStats().incMinaQueuedEvent();
			if (buf.processingCompleted) {
				buf.processingCompleted = false;
				execute = true;
			} else {
				execute = false;
			}
		}

		if (execute) {
			if (logger.isDebugEnabled()) {
				logger.debug("Launching thread for "
						+ session.getRemoteAddress());
			}

			executor.execute(new ProcessEventsRunnable(buf));
		}
	}

	private static class SessionBuffer {
		private static final String KEY = SessionBuffer.class.getName()
				+ ".KEY";

		private static SessionBuffer getSessionBuffer(IoSession session) {
			synchronized (session) {
				SessionBuffer buf = (SessionBuffer) session.getAttribute(KEY);
				if (buf == null) {
					buf = new SessionBuffer(session);
					session.setAttribute(KEY, buf);
				}
				return buf;
			}
		}

		private final IoSession session;

		private final Queue<Event> eventQueue = new LinkedList<Event>();

		private boolean processingCompleted = true;

		private SessionBuffer(IoSession session) {
			this.session = session;
		}
	}

	protected static class EventType {
		public static final EventType OPENED = new EventType("OPENED");

		public static final EventType CLOSED = new EventType("CLOSED");

		public static final EventType READ = new EventType("READ");

		public static final EventType WRITTEN = new EventType("WRITTEN");

		public static final EventType RECEIVED = new EventType("RECEIVED");

		public static final EventType SENT = new EventType("SENT");

		public static final EventType IDLE = new EventType("IDLE");

		public static final EventType EXCEPTION = new EventType("EXCEPTION");

		private final String value;

		private EventType(String value) {
			this.value = value;
		}

		public String toString() {
			return value;
		}
	}

	protected static class Event {
		private final EventType type;

		private final NextFilter nextFilter;

		private final Object data;

		Event(EventType type, NextFilter nextFilter, Object data) {
			this.type = type;
			this.nextFilter = nextFilter;
			this.data = data;
		}

		public Object getData() {
			return data;
		}

		public NextFilter getNextFilter() {
			return nextFilter;
		}

		public EventType getType() {
			return type;
		}
	}

	public void sessionCreated(NextFilter nextFilter, IoSession session) {
		nextFilter.sessionCreated(session);
	}

	public void sessionOpened(NextFilter nextFilter, IoSession session) {
		fireEvent(nextFilter, session, EventType.OPENED, null);
	}

	public void sessionClosed(NextFilter nextFilter, IoSession session) {
		fireEvent(nextFilter, session, EventType.CLOSED, null);
	}

	public void sessionIdle(NextFilter nextFilter, IoSession session,
			IdleStatus status) {
		fireEvent(nextFilter, session, EventType.IDLE, status);
	}

	public void exceptionCaught(NextFilter nextFilter, IoSession session,
			Throwable cause) {
		fireEvent(nextFilter, session, EventType.EXCEPTION, cause);
	}

	public void messageReceived(NextFilter nextFilter, IoSession session,
			Object message) {
		fireEvent(nextFilter, session, EventType.RECEIVED, message);
	}

	public void messageSent(NextFilter nextFilter, IoSession session,
			Object message) {
		fireEvent(nextFilter, session, EventType.SENT, message);
	}

	protected void processEvent(NextFilter nextFilter, IoSession session,
			EventType type, Object data) {
		if (type == EventType.RECEIVED) {
			nextFilter.messageReceived(session, data);
		} else if (type == EventType.SENT) {
			nextFilter.messageSent(session, (WriteRequest) data);
		} else if (type == EventType.EXCEPTION) {
			nextFilter.exceptionCaught(session, (Throwable) data);
		} else if (type == EventType.IDLE) {
			nextFilter.sessionIdle(session, (IdleStatus) data);
		} else if (type == EventType.OPENED) {
			nextFilter.sessionOpened(session);
		} else if (type == EventType.CLOSED) {
			nextFilter.sessionClosed(session);
		}
	}

	public void filterWrite(NextFilter nextFilter, IoSession session,
			WriteRequest writeRequest) {
		nextFilter.filterWrite(session, writeRequest);
	}

	public void filterClose(NextFilter nextFilter, IoSession session)
			throws Exception {
		nextFilter.filterClose(session);
	}

	private class ProcessEventsRunnable implements Runnable {
		private final SessionBuffer buffer;

		ProcessEventsRunnable(SessionBuffer buffer) {
			this.buffer = buffer;
		}

		public void run() {
			while (true) {
				Event event;

				synchronized (buffer.eventQueue) {
					event = buffer.eventQueue.poll();
					UuahServer.getServerStats().decMinaQueuedEvent();
					if (event == null) {
						buffer.processingCompleted = true;
						break;
					}
				}

				processEvent(event.getNextFilter(), buffer.session, event
						.getType(), event.getData());
			}

			if (logger.isDebugEnabled()) {
				logger.debug("Exiting since queue is empty for "
						+ buffer.session.getRemoteAddress());
			}
		}
	}
}
